package cn.getech.data.development.bean

import cn.getech.data.development.bean.jdbc.{JDBCCollectionConfigBean, JDBCFieldBean, JDBCSqlParserResourceBean}
import cn.getech.data.development.enums.FlinkStreamHiveWriteMode
import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}

// 自定义sink 交互模型数据
object FlinkStreamJDBCObj {

  def stringToJsonArrayParse(str: String): Array[JDBCSqlParserResourceBean] = {
    if (str == null || str == "") return null
    jsonArrayParse(JSON.parseArray(str))
  }

  def jsonArrayParse(jdbcObj: JSONArray): Array[JDBCSqlParserResourceBean] = {
    if (jdbcObj == null) return null
    jdbcObj.toArray.map(x => {
      jsonParse(JSON.parseObject(x.toString))
    })
  }

  def jsonParse(jdbcObj: JSONObject): JDBCSqlParserResourceBean = {
    val jdbc = new JDBCSqlParserResourceBean
    jdbc.url = jdbcObj.getString("url")
    jdbc.typename = jdbcObj.getString("typename")
    jdbc.username = jdbcObj.getString("username")
    jdbc.password = jdbcObj.getString("password")
    jdbc.db = jdbcObj.getString("db")
    jdbc.tableName = jdbcObj.getString("tableName")
    jdbc.appendType = jdbcObj.getOrDefault("appendType", FlinkStreamHiveWriteMode.APPEND).toString
    jdbc.hiveWriteMode = FlinkStreamHiveWriteMode.withName(jdbc.appendType)
    jdbc.isPartitionTable = jdbcObj.getOrDefault("isPartitionTable", "false").toString.toBoolean
    jdbc.isDynamicPartition = jdbcObj.getOrDefault("isDynamicPartition", "false").toString.toBoolean
    jdbc.collectionConfig = jsonCollectionConfig(jdbcObj.getJSONObject("collectionConfig"))
    jdbc.fields = jsonArray(jdbcObj)
    jdbc.sinkTableName = jdbcObj.getString("sinkTableName")
    jdbc.delimitFormat = jdbcObj.getIntValue("delimitFormat")
    jdbc.outputFormat = jdbcObj.getString("outputFormat")
    jdbc
  }

  private def jsonCollectionConfig(obj: JSONObject): JDBCCollectionConfigBean = {
    if (obj == null || obj.isEmpty) return JDBCCollectionConfigBean("", "")
    JDBCCollectionConfigBean(
      obj.getString("collectionTimeName"),
      obj.getString("collectionTimeFormat"))
  }

  private def jsonArray(obj: JSONObject): Array[JDBCFieldBean] = {
    obj.getJSONArray("fields").toArray.map(x => {
      val jsonObj = JSON.parseObject(x.toString)
      val field_name = jsonObj.getString("field_name")
      val class_type = jsonObj.getString("class_type")
      val isPartition = jsonObj.getBooleanValue("isPartition")
      val partitionValue = jsonObj.getOrDefault("partitionValue", "").toString
      val parsePartitionPath = jsonObj.getOrDefault("parsePartitionPath", "").toString
      JDBCFieldBean(field_name, class_type, isPartition, parsePartitionPath, partitionValue)
    })
  }
}
